package com.tom.study.example.selector;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.Channel;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicInteger;

public class SelectorThreadGroup {

    SelectorThread[] threads;

    ServerSocketChannel server = null;

    AtomicInteger xid = new AtomicInteger(0);

    SelectorThreadGroup stg = this;

    String type;

    public SelectorThreadGroup(String type,int threadNum) {
        this.threads = new SelectorThread[threadNum];
        for (int i = 0; i < threadNum; i++) {
            threads[i] = new SelectorThread(this);
            new Thread(threads[i], type + "Thread-" + i).start();
        }
    }

    public void bind(int port) {
        try {
            server = ServerSocketChannel.open();
            server.configureBlocking(false);
            server.bind(new InetSocketAddress(port));

            nextSelector(server);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void nextSelector(Channel c) {
        try {
            if (c instanceof ServerSocketChannel) {
                SelectorThread st = next();
                st.lbq.put(c);
                st.setWorker(stg);
                st.selector.wakeup();
            }else if(c instanceof SocketChannel){
                SelectorThread st = nextWorker();
                st.lbq.put(c);
                st.selector.wakeup();
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

    }

    private SelectorThread nextWorker() {
        int index = xid.incrementAndGet() % stg.threads.length;
        return stg.threads[index];
    }

    private SelectorThread next() {
        int index = xid.incrementAndGet() % threads.length;
        return threads[index];
    }

    public void setWorker(SelectorThreadGroup worker) {
        this.stg = worker;
    }
}
